import tushare as ts
import datetime as dt
import pandas as pd
from datetime import datetime
from django.conf import settings
import logging,time,asyncio
from asgiref.sync import sync_to_async
from .models import *
from .constants import *
from django.db.models import Max
from django.db import transaction


logger = logging.getLogger(__name__)

# 创建异步的数据库操作函数
# update_or_create_stock = sync_to_async(StockBasic.objects.update_or_create)
# update_or_create_trade_date = sync_to_async(TradeDate.objects.update_or_create)
# update_or_create_daily_data = sync_to_async(DailyData.objects.update_or_create)
# update_or_create_daily_basic = sync_to_async(DailyBasic.objects.update_or_create)
# update_or_create_ths_plate = sync_to_async(ThsPlate.objects.update_or_create)
# update_or_create_ths_stock_plate = sync_to_async(ThsStockPlate.objects.update_or_create)
update_data_record = sync_to_async(DataUpdateRecord.update_record)

# 定义一个同步函数来获取所有股票列表（排除ST股票）
def get_stocks_list():
    # 使用Q对象进行复杂查询，排除名称中包含"ST"的股票
    # from django.db.models import Q
    # return list(StockBasic.objects.exclude(
    #     Q(name__contains='ST') | Q(name__contains='st') | Q(name__contains='St')
    # ).all())
    return list(StockBasic.objects.all())

# 将同步函数转换为异步函数
get_all_stocks = sync_to_async(get_stocks_list)

# 获取数据类型的最后更新日期,返回值类型为datetime.date对象
def get_last_update_date(data_type):
    """
    获取指定数据类型的最后更新日期
    :param data_type: 数据类型，来自DATA_TYPES常量
    :return: 最后更新日期，如果没有记录则返回START_KDATA_DATE
    """
    
    if data_type == 'index_weekly':
        start_date =  START_KDATA_DATE_WEEK
    elif data_type == 'index_monthly':
        start_date = START_KDATA_DATE_MONTH
    else:
        start_date = START_KDATA_DATE
    try:
        record = DataUpdateRecord.objects.filter(data_type=data_type).order_by('-last_update_date').first()
        return record.last_update_date if record else start_date
    except Exception as e:
        logger.error(f"Error getting last update date for {data_type}: {str(e)}")
        return start_date

# 计算两个日期之间的交易日数量
def count_trading_days(start_date, end_date):
    """
    计算两个日期之间的交易日数量
    :param start_date: 开始日期，datetime.date对象或START_KDATA_DATE常量
    :param end_date: 结束日期，datetime.date对象
    :return: 交易日数量
    """
    if not start_date or not end_date:
        return 0
    
    # 如果start_date是START_KDATA_DATE常量，直接使用它（因为它已经是datetime.date对象）
    if start_date == START_KDATA_DATE:
        start_date = START_KDATA_DATE
    
    if start_date > end_date:#起始日期和结束日期相同时可能为一天
        return 0
    
    try:
        # 将日期转换为字符串格式YYYYMMDD
        start_str = start_date.strftime('%Y%m%d')
        end_str = end_date.strftime('%Y%m%d')
        
        # 查询交易日历
        trading_days = TradeDate.objects.filter(
            cal_date__gte=start_str,
            cal_date__lte=end_str,
            is_open=True
        ).count()
        
        return trading_days
    except Exception as e:
        logger.error(f"Error counting trading days: {str(e)}")
        return 0

# 将同步函数转换为异步函数
get_last_update_date_async = sync_to_async(get_last_update_date)
count_trading_days_async = sync_to_async(count_trading_days)
#完成所有的数据更新
#为了避免tushareapi接口过于庞大，将tushareapi接口拆分成多个文件
#本接口包括了tushare接口的基础数据、行情数据、指数专题数据
class TushareAPI:
    """Tushare API 交互类"""
    
    def __init__(self, token=None):
        """
        初始化 Tushare API
        :param token: Tushare API token，如果未提供则从环境变量获取
        """
        self.token = token or getattr(settings, 'TUSHARE_TOKEN', None)
        if not self.token:
            raise ValueError("Tushare API token is required")
        
        # 初始化 Tushare API
        ts.set_token(self.token)
        self.pro = ts.pro_api()
        self.plate_to_create = []#当前更新的板块列表，作为全局变量用于更新板块的个股关系数据

    async def update_stock_basic(self):
        """
        更新股票基本信息（排除ST股票）
        :return: 更新的记录数
        """
        try:
            # 获取股票列表
            df = self.pro.stock_basic(
                exchange='',
                list_status='L',
                fields='ts_code,symbol,name,area,industry,market,list_date,is_hs'
            )
            
            if df.empty:
                return 0
            
            # 直接将DataFrame转换为模型实例列表
            @sync_to_async
            def bulk_create_stock_data(df):
                with transaction.atomic():
                    # 获取现有的ts_code列表
                    existing_ts_codes = set(StockBasic.objects.values_list('ts_code', flat=True))
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        stock = StockBasic(
                            ts_code=row['ts_code'],
                            symbol=row['symbol'],
                            name=row['name'],
                            area=row['area'],
                            industry=row['industry'],
                            market=row['market'],
                            list_date=row['list_date'],
                            is_hs=row['is_hs']
                        )
                        
                        if row['ts_code'] in existing_ts_codes:
                            to_update.append(stock)
                        else:
                            to_create.append(stock)
                    
                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        StockBasic.objects.bulk_create(to_create, batch_size=1000)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        # 使用bulk_update更新现有记录
                        StockBasic.objects.bulk_update(
                            to_update, 
                            ['symbol', 'name', 'area', 'industry', 'market', 'list_date', 'is_hs'],
                            batch_size=1000
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建和更新
            count = await bulk_create_stock_data(df)
            

              # 获取股票公司信息列表
            df = self.pro.stock_company(
                fields='ts_code,com_name,manager,main_business,business_scope'
            )

            # 清空StockCompany表中的原有数据并存入新数据
            @sync_to_async
            def bulk_create_stock_company(df):
                with transaction.atomic():
                    # 清空表
                    StockCompany.objects.all().delete()
                    
                    # 准备要创建的记录
                    to_create = []
                    for _, row in df.iterrows():
                        stock_company = StockCompany(
                            ts_code=row['ts_code'],
                            com_name=row['com_name'],
                            manager=row['manager'],
                            main_business=row['main_business'],
                            business_scope=row['business_scope'],
                            update_date=dt.date.today().strftime('%Y%m%d')
                        )
                        to_create.append(stock_company)
                    
                    # 批量创建新记录
                    if to_create:
                        StockCompany.objects.bulk_create(to_create, batch_size=1000)
                    return len(to_create)
            
            # 执行批量创建
            company_count = await bulk_create_stock_company(df)
            logger.info(f"Successfully updated {company_count} stock company records")
            
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['STOCK_BASIC'],
                dt.date.today()
            )    
            logger.info(f"Successfully updated {count} stock basic records")
            return count
            
        except Exception as e:
            logger.error(f"Error updating stock basic data: {str(e)}")
            raise

    async def update_trade_calendar(self, start_date=None, end_date=None):
        """
        更新交易日历
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :return: 更新的记录数
        """
        try:
            # 如果未提供日期，则默认获取当年的数据
            if not start_date:
                start_date = datetime.now().strftime('%Y0101')
            if not end_date:
                end_date = datetime.now().strftime('%Y1231')

            # 获取交易日历
            df = self.pro.trade_cal(
                exchange='SSE',
                start_date=start_date,
                end_date=end_date,
                fields='exchange,cal_date,is_open,pretrade_date'
            )
            
            if df.empty:
                return 0
            
            # 直接将DataFrame转换为模型实例列表
            @sync_to_async
            def bulk_create_trade_calendar(df):
                with transaction.atomic():
                    # 获取现有的记录键值对
                    existing_records = {
                        (record.exchange, record.cal_date): record 
                        for record in TradeDate.objects.filter(
                            exchange='SSE',
                            cal_date__gte=start_date,
                            cal_date__lte=end_date
                        )
                    }
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        record_key = (row['exchange'], row['cal_date'])
                        trade_date = TradeDate(
                            exchange=row['exchange'],
                            cal_date=row['cal_date'],
                            is_open=row['is_open'] == 1,
                            pretrade_date=row['pretrade_date']
                        )
                        
                        if record_key in existing_records:
                            # 更新现有记录
                            existing_record = existing_records[record_key]
                            existing_record.is_open = row['is_open'] == 1
                            existing_record.pretrade_date = row['pretrade_date']
                            to_update.append(existing_record)
                        else:
                            # 创建新记录
                            to_create.append(trade_date)
                    
                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        TradeDate.objects.bulk_create(to_create, batch_size=1000)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        TradeDate.objects.bulk_update(
                            to_update, 
                            ['is_open', 'pretrade_date'],
                            batch_size=1000
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建和更新
            count = await bulk_create_trade_calendar(df)
            
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['TRADE_CAL'],
                dt.date.today()
            )    
            logger.info(f"Successfully updated {count} trade calendar records")
            return count
            
        except Exception as e:
            logger.error(f"Error updating trade calendar: {str(e)}")
            raise

    async def update_batch_daily_data(self, ts_codes, start_date=None, end_date=None):
        """
        批量更新多个股票的日线数据
        :param ts_codes: 股票代码列表（Tushare格式）
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :return: 更新的记录数
        """
        try:
            # 将股票代码列表转换为字符串，用逗号分隔
            ts_codes_str = ','.join(ts_codes)
            
            # 获取日线数据
            df = self.pro.daily(
                ts_code=ts_codes_str,
                start_date=start_date,
                end_date=end_date
            )
            
            if df.empty:
                return 0
            
            # 直接将DataFrame转换为模型实例列表
            @sync_to_async
            def bulk_create_daily_data(df):
                with transaction.atomic():
                    # 创建唯一键列表，用于避免重复数据
                    unique_keys = set()
                    for _, row in df.iterrows():
                        unique_keys.add((row['ts_code'], row['trade_date']))
                    
                    # 获取现有的记录
                    existing_records = {
                        (record.ts_code, record.trade_date): record 
                        for record in DailyData.objects.filter(
                            ts_code__in=df['ts_code'].unique(),
                            trade_date__gte=start_date,
                            trade_date__lte=end_date
                        )
                    }
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        record_key = (row['ts_code'], row['trade_date'])
                        
                        # 避免处理重复数据
                        if record_key in unique_keys:
                            unique_keys.remove(record_key)  # 移除已处理的键
                            
                            daily_data = DailyData(
                                ts_code=row['ts_code'],
                                trade_date=row['trade_date'],
                                open=row['open'],
                                high=row['high'],
                                low=row['low'],
                                close=row['close'],
                                pre_close=row['pre_close'],
                                change=row['change'],
                                pct_chg=row['pct_chg'],
                                vol=row['vol'],
                                amount=row['amount']
                            )
                            
                            if record_key in existing_records:
                                # 更新现有记录
                                existing_record = existing_records[record_key]
                                existing_record.open = row['open']
                                existing_record.high = row['high']
                                existing_record.low = row['low']
                                existing_record.close = row['close']
                                existing_record.pre_close = row['pre_close']
                                existing_record.change = row['change']
                                existing_record.pct_chg = row['pct_chg']
                                existing_record.vol = row['vol']
                                existing_record.amount = row['amount']
                                to_update.append(existing_record)
                            else:
                                # 创建新记录
                                to_create.append(daily_data)
                    
                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        DailyData.objects.bulk_create(to_create, batch_size=6000)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        DailyData.objects.bulk_update(
                            to_update, 
                            ['open', 'high', 'low', 'close', 'pre_close', 'change', 'pct_chg', 'vol', 'amount'],
                            batch_size=6000
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建和更新
            count = await bulk_create_daily_data(df)
            return count
            
        except Exception as e:
            logger.error(f"Error updating batch daily data: {str(e)}")
            raise
    #按日更新所有股票的日线数据
    async def batch_update_daily_data(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新所有股票的日线数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 更新的记录数
        """
        logger.info(f'Starting batch update of all daily data from {start_date} to {end_date}')
        total_count = 0
        try:
            # 获取所有股票列表
            stocks = await get_all_stocks()
            
            # 解析日期
            try:
                if start_date and end_date:
                    if '-' in start_date:  # 格式为YYYY-MM-DD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                    
                    if '-' in end_date:  # 格式为YYYY-MM-DD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
                else:
                    # 如果没有指定日期范围，默认更新最近10个交易日
                    end_date_obj = dt.date.today()
                    start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日
            except Exception as e:
                logger.error(f"Error parsing dates: {str(e)}")
                raise
            
            # 获取Tushare API限制
            max_data_count = TUSHARE_API_LIMIT.get('daily_k_max_count', 6000)  # 每次调用最多返回的数据条数
            max_stock_num = TUSHARE_API_LIMIT.get('daily_k_max_num', 600)      # 每次调用最多的股票数量
            
            # 计算每批次股票可以获取的最大交易日数
            max_days_per_batch = max_data_count // max_stock_num  # 通常是10天
            
            # 获取所有交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))
            
            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0
            
            # 将股票列表分成多个批次
            stock_batches = [stocks[i:i + max_stock_num] for i in range(0, len(stocks), max_stock_num)]
            total_batches = len(stock_batches)
            
            # 将交易日列表分成多个时间段
            date_batches = [trade_dates[i:i + max_days_per_batch] for i in range(0, len(trade_dates), max_days_per_batch)]
            total_date_batches = len(date_batches)
            
            batch_count = 0
            total_batch_count = total_batches * total_date_batches
            
            # 对每个股票批次和每个时间段进行处理
            for stock_batch_index, stock_batch in enumerate(stock_batches, 1):
                ts_codes = [stock.ts_code for stock in stock_batch]
                
                for date_batch_index, date_batch in enumerate(date_batches, 1):
                    batch_count += 1
                    
                    # 发送进度更新
                    if websocket:
                        message = (f"正在更新第 {stock_batch_index}/{total_batches} 批次, "
                                 f"第 {date_batch_index}/{total_date_batches} 个时间段 "
                                 f"({len(date_batch)}个交易日, {len(stock_batch)}只股票日K线数据)")
                        try:
                            await websocket.send_progress(
                                current=batch_count,
                                total=total_batch_count,
                                message=message
                            )
                        except Exception as e:
                            logger.error(f"WebSocket发送失败: {str(e)}")
                    
                    try:
                        # 批量更新数据
                        count = await self.update_batch_daily_data(
                            ts_codes=ts_codes,
                            start_date=date_batch[0],
                            end_date=date_batch[-1]
                        )
                        total_count += count
                    except Exception as e:
                        logger.error(f"Error updating batch {batch_count}/{total_batch_count}: {str(e)}")
                        continue
            
            # 更新数据更新记录
            if end_date:
                # 根据end_date的格式进行不同的处理
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                end_date_obj = dt.date.today()
                
            await update_data_record(
                DATA_TYPES['DAILY'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {total_count} daily records in total")
            return total_count
            
        except Exception as e:
            logger.error(f"Error in batch update: {str(e)}")
            raise

    

    async def batch_update_stock_weekly(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新所有股票的周线数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 更新的记录数
        """
        logger.info(f'Starting batch update of all weekly data from {start_date} to {end_date}')
        total_count = 0
        try:
            # 获取所有股票列表
            # stocks = await get_all_stocks()
            # ts_codes = [stock.ts_code for stock in stocks]
            
            # 解析日期
            try:
                if start_date and end_date:
                    if '-' in start_date:  # 格式为YYYY-MM-DD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                    
                    if '-' in end_date:  # 格式为YYYY-MM-DD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
                else:
                    # 如果没有指定日期范围，默认更新最近10个周
                    end_date_obj = dt.date.today()
                    start_date_obj = end_date_obj - dt.timedelta(days=70)  # 多取几天以确保包含10个周
            except Exception as e:
                logger.error(f"Error parsing dates: {str(e)}")
                raise
            
            # 获取所有周五日期列表
            friday_dates = []
            current_date = start_date_obj
            while current_date <= end_date_obj:
                if current_date.weekday() == 4:  # 周五
                    friday_dates.append(current_date.strftime('%Y%m%d'))
                current_date += dt.timedelta(days=1)
            
            if not friday_dates:
                logger.warning("No Friday dates found in the specified date range")
                return 0
            
            # 发送进度更新
            if websocket:
                message = f"正在更新所有股票的周K线数据"
                try:
                    await websocket.send_progress(
                        current=1,
                        total=1,
                        message=message
                    )
                except Exception as e:
                    logger.error(f"WebSocket发送失败: {str(e)}")
            
            try:
                # 批量更新数据
                count = await self.update_batch_weekly_data(
                    trade_dates=friday_dates
                )
                total_count += count
            except Exception as e:
                logger.error(f"Error updating weekly data: {str(e)}")
                raise
            
            # 更新数据更新记录
            if end_date:
                # 根据end_date的格式进行不同的处理
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                end_date_obj = dt.date.today()
                
            await update_data_record(
                DATA_TYPES['STOCK_WEEKLY'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {total_count} weekly records in total")
            return total_count
            
        except Exception as e:
            logger.error(f"Error in batch update: {str(e)}")
            raise
    #按周更新所有股票的日线数据
    async def update_batch_weekly_data(self, trade_dates):
        """
        批量更新多个股票的周线数据
        :param ts_codes: 股票代码列表（Tushare格式）
        :param trade_dates: 交易日期列表，格式：YYYYMMDD
        :return: 更新的记录数
        """

        # 直接将DataFrame转换为模型实例列表
        @sync_to_async
        def bulk_create_weekly_data(df):
            with transaction.atomic():
                # 创建唯一键列表，用于避免重复数据
                unique_keys = set()
                for _, row in df.iterrows():
                    unique_keys.add((row['ts_code'], row['trade_date']))
                
                # 获取现有的记录
                existing_records = {
                    (record.ts_code, record.trade_date): record 
                    for record in StockKDataWeekly.objects.filter(
                        ts_code__in=df['ts_code'].unique(),
                        trade_date__in=trade_dates
                    )
                }
                
                # 准备要创建和更新的记录
                to_create = []
                to_update = []
                
                for _, row in df.iterrows():
                    record_key = (row['ts_code'], row['trade_date'])
                    
                    # 避免处理重复数据
                    if record_key in unique_keys:
                        unique_keys.remove(record_key)  # 移除已处理的键
                        
                        weekly_data = StockKDataWeekly(
                            ts_code=row['ts_code'],
                            trade_date=row['trade_date'],
                            open=row['open'],
                            high=row['high'],
                            low=row['low'],
                            close=row['close'],
                            pre_close=row['pre_close'],
                            change=row['change'],
                            pct_chg=row['pct_chg'],
                            vol=row['vol'],
                            amount=row['amount']
                        )
                        
                        if record_key in existing_records:
                            # 更新现有记录
                            existing_record = existing_records[record_key]
                            existing_record.open = row['open']
                            existing_record.high = row['high']
                            existing_record.low = row['low']
                            existing_record.close = row['close']
                            existing_record.pre_close = row['pre_close']
                            existing_record.change = row['change']
                            existing_record.pct_chg = row['pct_chg']
                            existing_record.vol = row['vol']
                            existing_record.amount = row['amount']
                            to_update.append(existing_record)
                        else:
                            # 创建新记录
                            to_create.append(weekly_data)
                
                # 批量创建新记录
                created_count = 0
                if to_create:
                    StockKDataWeekly.objects.bulk_create(to_create, batch_size=6000)
                    created_count = len(to_create)
                
                # 批量更新现有记录
                updated_count = 0
                if to_update:
                    StockKDataWeekly.objects.bulk_update(
                        to_update, 
                        ['open', 'high', 'low', 'close', 'pre_close', 'change', 'pct_chg', 'vol', 'amount'],
                        batch_size=6000
                    )
                    updated_count = len(to_update)
                
                return created_count + updated_count
        try:
            # 获取周线数据.每次写入全部个股数据，以一周写一次。也不是很多。避免数据过大。
            for trade_date in trade_dates:
                df = self.pro.weekly(trade_date=trade_date)
                # 执行批量创建和更新
                count = await bulk_create_weekly_data(df)
            return count
            
        except Exception as e:
            logger.error(f"Error updating batch weekly data: {str(e)}")
            raise
    #按月更新所有股票的日线数据
    async def batch_update_stock_monthly(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新所有股票的月线数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 更新的记录数
        """
        logger.info(f'Starting batch update of all monthly data from {start_date} to {end_date}')
        total_count = 0
        try:
            
            # 解析日期
            try:
                if start_date and end_date:
                    if '-' in start_date:  # 格式为YYYY-MM-DD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                    
                    if '-' in end_date:  # 格式为YYYY-MM-DD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
                else:
                    # 如果没有指定日期范围，默认更新最近12个月
                    end_date_obj = dt.date.today()
                    start_date_obj = end_date_obj - dt.timedelta(days=365)  # 多取几天以确保包含12个月
            except Exception as e:
                logger.error(f"Error parsing dates: {str(e)}")
                raise
            
            # 获取所有月末日期列表
            month_end_dates = []
            current_date = start_date_obj
            while current_date <= end_date_obj:
                # 获取当前月的最后一天
                next_month = current_date.replace(day=28) + dt.timedelta(days=4)  # 确保进入下个月
                month_end = next_month - dt.timedelta(days=next_month.day)
                month_end_dates.append(month_end.strftime('%Y%m%d'))
                current_date = next_month
            
            if not month_end_dates:
                logger.warning("No month end dates found in the specified date range")
                return 0
            
            # 发送进度更新
            if websocket:
                message = f"正在更新所有股票的月K线数据"
                try:
                    await websocket.send_progress(
                        current=0,
                        total=len(month_end_dates),
                        message=message
                    )
                except Exception as e:
                    logger.error(f"WebSocket发送失败: {str(e)}")
            
            # 遍历月末日期，获取数据并存储
            for trade_date in month_end_dates:
                try:
                    # 调用Tushare接口获取数据
                    df = self.pro.monthly(trade_date=trade_date)
                    if df is not None and not df.empty:
                        # 将数据存储到数据库
                        @sync_to_async
                        def bulk_create_monthly_data(df):
                            with transaction.atomic():
                                # 获取现有的记录
                                existing_records = {
                                    (record.ts_code, record.trade_date): record 
                                    for record in StockKDataMonthly.objects.filter(
                                        ts_code__in=df['ts_code'].unique(),
                                        trade_date=trade_date
                                    )
                                }
                                
                                # 准备要创建和更新的记录
                                to_create = []
                                to_update = []
                                
                                for _, row in df.iterrows():
                                    record_key = (row['ts_code'], trade_date)
                                    monthly_data = StockKDataMonthly(
                                        ts_code=row['ts_code'],
                                        trade_date=trade_date,
                                        open=row['open'],
                                        high=row['high'],
                                        low=row['low'],
                                        close=row['close'],
                                        pre_close=row['pre_close'],
                                        change=row['change'],
                                        pct_chg=row['pct_chg'],
                                        vol=row['vol'],
                                        amount=row['amount']
                                    )
                                    
                                    if record_key in existing_records:
                                        # 更新现有记录
                                        existing_record = existing_records[record_key]
                                        existing_record.open = row['open']
                                        existing_record.high = row['high']
                                        existing_record.low = row['low']
                                        existing_record.close = row['close']
                                        existing_record.pre_close = row['pre_close']
                                        existing_record.change = row['change']
                                        existing_record.pct_chg = row['pct_chg']
                                        existing_record.vol = row['vol']
                                        existing_record.amount = row['amount']
                                        to_update.append(existing_record)
                                    else:
                                        # 创建新记录
                                        to_create.append(monthly_data)
                                
                                # 批量创建新记录
                                created_count = 0
                                if to_create:
                                    StockKDataMonthly.objects.bulk_create(to_create, batch_size=1000)
                                    created_count = len(to_create)
                                
                                # 批量更新现有记录
                                updated_count = 0
                                if to_update:
                                    StockKDataMonthly.objects.bulk_update(
                                        to_update, 
                                        ['open', 'high', 'low', 'close', 'pre_close', 'change', 'pct_chg', 'vol', 'amount'],
                                        batch_size=1000
                                    )
                                    updated_count = len(to_update)
                                
                                return created_count + updated_count
                        
                        # 执行批量创建和更新
                        count = await bulk_create_monthly_data(df)
                        total_count += count
                except Exception as e:
                    logger.error(f"Error updating monthly data for {trade_date}: {str(e)}")
                    continue
            
            # 更新数据更新记录
            if end_date:
                # 根据end_date的格式进行不同的处理
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                end_date_obj = dt.date.today()
                
            await update_data_record(
                DATA_TYPES['STOCK_MONTHLY'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {total_count} monthly records in total")
            return total_count
            
        except Exception as e:
            logger.error(f"Error in batch update: {str(e)}")
            raise




    async def update_st_stock(self, start_date=None, end_date=None):
        try:
            # 如果未提供日期，则默认获取最近10个交易日的数据
            if not start_date or not end_date:
                end_date = dt.date.today().strftime('%Y%m%d')
                start_date = (dt.date.today() - dt.timedelta(days=10)).strftime('%Y%m%d')

            # 获取日期范围内的所有交易日
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date,
                cal_date__lte=end_date,
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            # 遍历每个交易日，获取ST股票数据
            all_st_data = []
            loop_count = 0
            start_time = time.time()
            stock_st_max_count_in_minute = TUSHARE_API_LIMIT.get('st_list_max_in_minute', 200)
            for trade_date in trade_dates:
                loop_count += 1
                print(f'trade_date: {trade_date}')
                try:
                    # 调用Tushare接口获取ST股票数据
                    df = self.pro.stock_st(
                        trade_date=trade_date,
                        fields='ts_code,name,trade_date,type,type_name'
                    )
                    if not df.empty:
                        df['trade_date'] = trade_date
                        all_st_data.append(df)
                except Exception as e:
                    logger.error(f"Error fetching ST stock data for date {trade_date}: {str(e)}")
                    continue
                
                # 每200次循环检查一次时间
                if loop_count % stock_st_max_count_in_minute == 0:
                    elapsed_time = time.time() - start_time
                    if elapsed_time < 60:  # 如果运行时间不足1分钟
                        sleep_time = 61 - elapsed_time  # 等待到1分01秒
                        logger.info(f"Waiting for {sleep_time:.2f} seconds to comply with rate limits")
                        time.sleep(sleep_time)
                    start_time = time.time()  # 重置计时器

            if not all_st_data:
                logger.warning("No ST stock data found in the specified date range")
                return 0

            # 合并所有数据
            combined_df = pd.concat(all_st_data)

            # 批量存储到数据库
            @sync_to_async
            def bulk_create_st_data(df):
                with transaction.atomic():
                    # 获取现有的记录
                    existing_records = {
                        (record.ts_code, record.trade_date): record
                        for record in Stock_ST.objects.filter(
                            trade_date__gte=start_date,
                            trade_date__lte=end_date
                        )
                    }

                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []

                    for _, row in df.iterrows():
                        record_key = (row['ts_code'], row['trade_date'])
                        st_data = Stock_ST(
                            ts_code=row['ts_code'],
                            name=row['name'],
                            trade_date=row['trade_date'],
                            type='ST',  # 默认类型为ST
                            type_name='ST股票'  # 默认类型名称为ST股票
                        )

                        if record_key in existing_records:
                            # 更新现有记录
                            existing_record = existing_records[record_key]
                            existing_record.name = row['name']
                            existing_record.type = 'ST'
                            existing_record.type_name = 'ST股票'
                            to_update.append(existing_record)
                        else:
                            # 创建新记录
                            to_create.append(st_data)

                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        Stock_ST.objects.bulk_create(to_create, batch_size=1000)
                        created_count = len(to_create)

                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        Stock_ST.objects.bulk_update(
                            to_update,
                            ['name', 'type', 'type_name'],
                            batch_size=1000
                        )
                        updated_count = len(to_update)

                    return created_count + updated_count

            # 执行批量创建和更新
            count = await bulk_create_st_data(combined_df)

            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['STOCK_ST'],
                dt.date.today()
            )

            logger.info(f"Successfully updated {count} ST stock records")
            return count

        except Exception as e:
            logger.error(f"Error updating ST stock data: {str(e)}")
            raise
    async def update_ipo(self, start_date=None, end_date=None):
    # """
    # 更新IPO股票列表数据
    # :param start_date: 开始日期，格式：YYYYMMDD
    # :param end_date: 结束日期，格式：YYYYMMDD
    # :return: 更新的记录数
    # """
        try:
            # 获取IPO股票列表数据
            df = self.pro.new_share(
                start_date=start_date,
                end_date=end_date
            )

            if df.empty:
                return 0

            # 转换为字典列表
            records = df.to_dict('records')

            # 批量写入数据库
            @sync_to_async
            def bulk_create_ipo_data(records):
                with transaction.atomic():
                    # 使用 bulk_create 避免重复数据
                    objs = []
                    for record in records:
                        # 检查是否已存在
                        if not IPO_NewShare.objects.filter(
                            ts_code=record['ts_code'],
                            ipo_date=record['ipo_date']
                        ).exists():
                            objs.append(IPO_NewShare(**record))

                    if objs:
                        IPO_NewShare.objects.bulk_create(objs)
                        return len(objs)
                    return 0

            # 执行批量创建
            count = await bulk_create_ipo_data(records)

            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['IPO_NEW'],
                dt.date.today()
            )
            logger.info(f"Successfully updated {count} IPO records")
            return count

        except Exception as e:
            logger.error(f"Error updating IPO data: {str(e)}")
            raise
    async def update_batch_daily_basic(self, ts_codes, start_date=None, end_date=None):
        """
        批量更新多个股票的每日指标数据
        :param ts_codes: 股票代码列表（Tushare格式）
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :return: 更新的记录数
        """
        try:
            # 将股票代码列表转换为字符串，用逗号分隔
            ts_codes_str = ','.join(ts_codes)
            
            # 获取每日指标数据
            df = self.pro.daily_basic(
                ts_code=ts_codes_str,
                start_date=start_date,
                end_date=end_date,
                fields='ts_code,trade_date,close,turnover_rate,turnover_rate_f,volume_ratio,pe,pe_ttm,pb,ps,ps_ttm,dv_ratio,dv_ttm,total_share,float_share,free_share,total_mv,circ_mv'
            )
            
            if df.empty:
                return 0
            
            # 直接将DataFrame转换为模型实例列表
            @sync_to_async
            def bulk_create_daily_basic(df):
                with transaction.atomic():
                    # 创建唯一键列表，用于避免重复数据
                    unique_keys = set()
                    for _, row in df.iterrows():
                        unique_keys.add((row['ts_code'], row['trade_date']))
                    
                    # 获取现有的记录
                    existing_records = {
                        (record.ts_code, record.trade_date): record 
                        for record in DailyBasic.objects.filter(
                            ts_code__in=df['ts_code'].unique(),
                            trade_date__gte=start_date,
                            trade_date__lte=end_date
                        )
                    }
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        record_key = (row['ts_code'], row['trade_date'])
                        
                        # 避免处理重复数据
                        if record_key in unique_keys:
                            unique_keys.remove(record_key)  # 移除已处理的键
                            
                            daily_basic = DailyBasic(
                                ts_code=row['ts_code'],
                                trade_date=row['trade_date'],
                                close=row['close'],
                                turnover_rate=row['turnover_rate'],
                                turnover_rate_f=row['turnover_rate_f'],
                                volume_ratio=row['volume_ratio'],
                                pe=row['pe'],
                                pe_ttm=row['pe_ttm'],
                                pb=row['pb'],
                                ps=row['ps'],
                                ps_ttm=row['ps_ttm'],
                                dv_ratio=row['dv_ratio'],
                                dv_ttm=row['dv_ttm'],
                                total_share=row['total_share'],
                                float_share=row['float_share'],
                                free_share=row['free_share'],
                                total_mv=row['total_mv'],
                                circ_mv=row['circ_mv']
                            )
                            
                            if record_key in existing_records:
                                # 更新现有记录
                                existing_record = existing_records[record_key]
                                existing_record.close = row['close']
                                existing_record.turnover_rate = row['turnover_rate']
                                existing_record.turnover_rate_f = row['turnover_rate_f']
                                existing_record.volume_ratio = row['volume_ratio']
                                existing_record.pe = row['pe']
                                existing_record.pe_ttm = row['pe_ttm']
                                existing_record.pb = row['pb']
                                existing_record.ps = row['ps']
                                existing_record.ps_ttm = row['ps_ttm']
                                existing_record.dv_ratio = row['dv_ratio']
                                existing_record.dv_ttm = row['dv_ttm']
                                existing_record.total_share = row['total_share']
                                existing_record.float_share = row['float_share']
                                existing_record.free_share = row['free_share']
                                existing_record.total_mv = row['total_mv']
                                existing_record.circ_mv = row['circ_mv']
                                to_update.append(existing_record)
                            else:
                                # 创建新记录
                                to_create.append(daily_basic)
                    
                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        DailyBasic.objects.bulk_create(to_create, batch_size=6000)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        DailyBasic.objects.bulk_update(
                            to_update, 
                            ['close', 'turnover_rate', 'turnover_rate_f', 'volume_ratio', 'pe', 'pe_ttm', 'pb', 'ps', 'ps_ttm', 'dv_ratio', 'dv_ttm', 'total_share', 'float_share', 'free_share', 'total_mv', 'circ_mv'],
                            batch_size=6000
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建和更新
            count = await bulk_create_daily_basic(df)
            return count
            
        except Exception as e:
            logger.error(f"Error updating batch daily basic data: {str(e)}")
            raise

    async def batch_update_daily_basic(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新所有股票的每日指标数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 更新的记录数
        """
        logger.info(f'Starting batch update of all daily basic data from {start_date} to {end_date}')
        total_count = 0
        try:
            # 获取所有股票列表
            stocks = await get_all_stocks()
            
            # 解析日期
            try:
                if start_date and end_date:
                    if '-' in start_date:  # 格式为YYYY-MM-DD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                    
                    if '-' in end_date:  # 格式为YYYY-MM-DD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
                else:
                    # 如果没有指定日期范围，默认更新最近10个交易日
                    end_date_obj = dt.date.today()
                    start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日
            except Exception as e:
                logger.error(f"Error parsing dates: {str(e)}")
                raise
            
            # 获取Tushare API限制
            max_data_count = TUSHARE_API_LIMIT.get('daily_k_max_count', 6000)  # 每次调用最多返回的数据条数
            max_stock_num = TUSHARE_API_LIMIT.get('daily_k_max_num', 600)      # 每次调用最多的股票数量
            
            # 计算每批次股票可以获取的最大交易日数
            max_days_per_batch = max_data_count // max_stock_num  # 通常是10天
            
            # 获取所有交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))
            
            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0
            
            # 将股票列表分成多个批次
            stock_batches = [stocks[i:i + max_stock_num] for i in range(0, len(stocks), max_stock_num)]
            total_batches = len(stock_batches)
            
            # 将交易日列表分成多个时间段
            date_batches = [trade_dates[i:i + max_days_per_batch] for i in range(0, len(trade_dates), max_days_per_batch)]
            total_date_batches = len(date_batches)
            
            batch_count = 0
            total_batch_count = total_batches * total_date_batches
            
            # 对每个股票批次和每个时间段进行处理
            for stock_batch_index, stock_batch in enumerate(stock_batches, 1):
                ts_codes = [stock.ts_code for stock in stock_batch]
                
                for date_batch_index, date_batch in enumerate(date_batches, 1):
                    batch_count += 1
                    
                    # 发送进度更新
                    if websocket:
                        message = (f"正在更新第 {stock_batch_index}/{total_batches} 批次股票的每日指标, "
                                 f"第 {date_batch_index}/{total_date_batches} 个时间段 "
                                 f"({len(date_batch)}个交易日, {len(stock_batch)}只股票)")
                        try:
                            await websocket.send_progress(
                                current=batch_count,
                                total=total_batch_count,
                                message=message
                            )
                        except Exception as e:
                            logger.error(f"WebSocket发送失败: {str(e)}")
                    
                    try:
                        # 批量更新数据
                        count = await self.update_batch_daily_basic(
                            ts_codes=ts_codes,
                            start_date=date_batch[0],
                            end_date=date_batch[-1]
                        )
                        total_count += count
                    except Exception as e:
                        logger.error(f"Error updating batch {batch_count}/{total_batch_count}: {str(e)}")
                        continue
            
            # 更新数据更新记录
            if end_date:
                # 根据end_date的格式进行不同的处理
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                end_date_obj = dt.date.today()
                
            await update_data_record(
                DATA_TYPES['DAILY_BASIC'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {total_count} daily basic records in total")
            return total_count
            
        except Exception as e:
            logger.error(f"Error in batch update: {str(e)}")
            raise

    
    async def update_batch_index_daily(self, ts_code, start_date=None, end_date=None):
        """
        更新单个指数的日线数据
        :param ts_code: 指数代码（Tushare格式）
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :return: 更新的记录数
        """
        try:
            # 获取指数日线数据
            df = self.pro.index_daily(
                ts_code=ts_code,
                start_date=start_date,
                end_date=end_date,
                fields='ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount'
            )
            
            # 获取指数每日指标数据
            df_basic = self.pro.index_dailybasic(
                ts_code=ts_code,
                start_date=start_date,
                end_date=end_date,
                fields='ts_code,trade_date,total_mv,float_mv,pe,pb,turnover_rate_f'
            )
            
            if not df.empty and not df_basic.empty:
                # 合并日线数据和每日指标数据
                df = pd.merge(df, df_basic, on=['ts_code', 'trade_date'], how='left')
            
            if df.empty:
                return 0
            
            # 直接将DataFrame转换为模型实例列表
            @sync_to_async
            def bulk_create_index_daily(df):
                with transaction.atomic():
                    # 创建唯一键列表，用于避免重复数据
                    unique_keys = set()
                    for _, row in df.iterrows():
                        unique_keys.add((row['ts_code'], row['trade_date']))
                    
                    # 获取现有的记录
                    existing_records = {
                        (record.ts_code, record.trade_date): record 
                        for record in IndexDaily.objects.filter(
                            ts_code__in=df['ts_code'].unique(),
                            trade_date__gte=start_date,
                            trade_date__lte=end_date
                        )
                    }
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        record_key = (row['ts_code'], row['trade_date'])
                        
                        # 避免处理重复数据
                        if record_key in unique_keys:
                            unique_keys.remove(record_key)  # 移除已处理的键
                            
                            index_daily = IndexDaily(
                                ts_code=row['ts_code'],
                                trade_date=row['trade_date'],
                                open=row['open'],
                                high=row['high'],
                                low=row['low'],
                                close=row['close'],
                                pre_close=row['pre_close'],
                                change=row['change'],
                                pct_chg=row['pct_chg'],
                                vol=row['vol'],
                                amount=row['amount'],
                                total_mv=row.get('total_mv'),
                                float_mv=row.get('float_mv'),
                                pe=row.get('pe'),
                                pb=row.get('pb'),
                                turnover_rate_f=row.get('turnover_rate_f')
                            )
                            
                            if record_key in existing_records:
                                # 更新现有记录
                                existing_record = existing_records[record_key]
                                for field in ['open', 'high', 'low', 'close', 'pre_close', 'change', 
                                            'pct_chg', 'vol', 'amount', 'total_mv', 'float_mv', 'pe', 'pb','turnover_rate_f']:
                                    setattr(existing_record, field, getattr(index_daily, field))
                                to_update.append(existing_record)
                            else:
                                # 创建新记录
                                to_create.append(index_daily)
                    
                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        IndexDaily.objects.bulk_create(to_create, batch_size=1000)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        IndexDaily.objects.bulk_update(
                            to_update, 
                            ['open', 'high', 'low', 'close', 'pre_close', 'change', 'pct_chg', 
                             'vol', 'amount', 'total_mv', 'float_mv', 'pe', 'pb', 'turnover_rate_f'],
                            batch_size=1000
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建和更新
            count = await bulk_create_index_daily(df)
            return count
            
        except Exception as e:
            logger.error(f"Error updating batch index daily data: {str(e)}")
            raise

    async def batch_update_index_daily(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新所有指数的日线数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 更新的记录数
        """
        logger.info(f'Starting batch update of all index daily data from {start_date} to {end_date}')
        total_count = 0
        try:
            # 使用常量中定义的主要指数列表
            ts_codes = [index['ts_code'] for index in MAIN_INDEXES]
            
            # 解析日期
            try:
                if start_date and end_date:
                    if '-' in start_date:  # 格式为YYYY-MM-DD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                    
                    if '-' in end_date:  # 格式为YYYY-MM-DD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
                else:
                    # 如果没有指定日期范围，默认更新最近10个交易日
                    end_date_obj = dt.date.today()
                    start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日
            except Exception as e:
                logger.error(f"Error parsing dates: {str(e)}")
                raise
            
            # 获取所有交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))
            
            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0
            
            # 逐个处理每个指数
            total_indexes = len(ts_codes)
            for index_num, ts_code in enumerate(ts_codes, 1):
                # 发送进度更新
                if websocket:
                    message = f"正在更新指数日线数据 ({index_num}/{total_indexes}): {ts_code}"
                    try:
                        await websocket.send_progress(
                            current=index_num,
                            total=total_indexes,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")
                
                try:
                    # 更新单个指数的数据
                    count = await self.update_batch_index_daily(
                        ts_code=ts_code,
                        start_date=trade_dates[0],
                        end_date=trade_dates[-1]
                    )
                    total_count += count
                    logger.info(f"Updated {count} records for index {ts_code}")
                except Exception as e:
                    logger.error(f"Error updating index {ts_code}: {str(e)}")
                    continue
            
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['INDEX_DAILY'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {total_count} index daily records in total")
            return total_count
            
        except Exception as e:
            logger.error(f"Error in batch update: {str(e)}")
            raise

    async def batch_update_index_weekly(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新所有指数的周线数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 更新的记录数
        """
        logger.info(f'Starting batch update of all index weekly data from {start_date} to {end_date}')
        total_count = 0
        try:
            # 使用常量中定义的主要指数列表
            ts_codes = [index['ts_code'] for index in MAIN_INDEXES]
            
            # 解析日期
            try:
                if start_date and end_date:
                    if '-' in start_date:  # 格式为YYYY-MM-DD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                    
                    if '-' in end_date:  # 格式为YYYY-MM-DD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
                else:
                    # 如果没有指定日期范围，默认更新最近3个月的数据
                    end_date_obj = dt.date.today()
                    start_date_obj = end_date_obj - dt.timedelta(days=90)  # 约3个月
            except Exception as e:
                logger.error(f"Error parsing dates: {str(e)}")
                raise
            
            # 将日期对象转换为字符串格式
            start_date_str = start_date_obj.strftime('%Y%m%d')
            end_date_str = end_date_obj.strftime('%Y%m%d')
            
            # 逐个处理每个指数
            total_indexes = len(ts_codes)
            for index_num, ts_code in enumerate(ts_codes, 1):
                # 发送进度更新
                if websocket:
                    message = f"正在更新指数周线数据 ({index_num}/{total_indexes}): {ts_code}"
                    try:
                        await websocket.send_progress(
                            current=index_num,
                            total=total_indexes,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")
                
                try:
                    # 更新单个指数的数据
                    count = await self.update_batch_index_weekly(
                        ts_code=ts_code,
                        start_date=start_date_str,
                        end_date=end_date_str
                    )
                    total_count += count
                    logger.info(f"Updated {count} weekly records for index {ts_code}")
                except Exception as e:
                    logger.error(f"Error updating weekly data for index {ts_code}: {str(e)}")
                    continue
            
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['INDEX_WEEKLY'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {total_count} index weekly records in total")
            return total_count
            
        except Exception as e:
            logger.error(f"Error in batch update of weekly data: {str(e)}")
            raise

    async def update_batch_index_weekly(self, ts_code, start_date=None, end_date=None):
        """
        更新单个指数的周线数据
        :param ts_code: 指数代码（Tushare格式）
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :return: 更新的记录数
        """
        try:
            # 获取指数周线数据
            df = self.pro.index_weekly(
                ts_code=ts_code,
                start_date=start_date,
                end_date=end_date,
                fields='ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount'
            )
            
            if df.empty:
                return 0
            
            # 直接将DataFrame转换为模型实例列表
            @sync_to_async
            def bulk_create_index_weekly(df):
                with transaction.atomic():
                    # 创建唯一键列表，用于避免重复数据
                    unique_keys = set()
                    for _, row in df.iterrows():
                        unique_keys.add((row['ts_code'], row['trade_date']))
                    
                    # 获取现有的记录
                    existing_records = {
                        (record.ts_code, record.trade_date): record 
                        for record in IndexWeekly.objects.filter(
                            ts_code__in=df['ts_code'].unique(),
                            trade_date__gte=start_date,
                            trade_date__lte=end_date
                        )
                    }
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        record_key = (row['ts_code'], row['trade_date'])
                        
                        # 避免处理重复数据
                        if record_key in unique_keys:
                            unique_keys.remove(record_key)  # 移除已处理的键
                            
                            index_weekly = IndexWeekly(
                                ts_code=row['ts_code'],
                                trade_date=row['trade_date'],
                                open=row['open'],
                                high=row['high'],
                                low=row['low'],
                                close=row['close'],
                                pre_close=row['pre_close'],
                                change=row['change'],
                                pct_chg=row['pct_chg'],
                                vol=row['vol'],
                                amount=row['amount']
                            )
                            
                            if record_key in existing_records:
                                # 更新现有记录
                                existing_record = existing_records[record_key]
                                for field in ['open', 'high', 'low', 'close', 'pre_close', 'change', 
                                            'pct_chg', 'vol', 'amount']:
                                    setattr(existing_record, field, getattr(index_weekly, field))
                                to_update.append(existing_record)
                            else:
                                # 创建新记录
                                to_create.append(index_weekly)
                    
                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        IndexWeekly.objects.bulk_create(to_create, batch_size=1000)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        IndexWeekly.objects.bulk_update(
                            to_update, 
                            ['open', 'high', 'low', 'close', 'pre_close', 'change', 'pct_chg', 
                             'vol', 'amount'],
                            batch_size=1000
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建和更新
            count = await bulk_create_index_weekly(df)
            return count
            
        except Exception as e:
            logger.error(f"Error updating batch index weekly data: {str(e)}")
            raise

    async def update_batch_index_monthly(self, ts_code, start_date=None, end_date=None):
        """
        更新单个指数的月线数据
        :param ts_code: 指数代码（Tushare格式）
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :return: 更新的记录数
        """
        try:
            # 获取指数月线数据
            df = self.pro.index_monthly(
                ts_code=ts_code,
                start_date=start_date,
                end_date=end_date,
                fields='ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount'
            )
            
            if df.empty:
                return 0
            
            # 直接将DataFrame转换为模型实例列表
            @sync_to_async
            def bulk_create_index_monthly(df):
                with transaction.atomic():
                    # 创建唯一键列表，用于避免重复数据
                    unique_keys = set()
                    for _, row in df.iterrows():
                        unique_keys.add((row['ts_code'], row['trade_date']))
                    
                    # 获取现有的记录
                    existing_records = {
                        (record.ts_code, record.trade_date): record 
                        for record in IndexMonthly.objects.filter(
                            ts_code__in=df['ts_code'].unique(),
                            trade_date__gte=start_date,
                            trade_date__lte=end_date
                        )
                    }
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        record_key = (row['ts_code'], row['trade_date'])
                        
                        # 避免处理重复数据
                        if record_key in unique_keys:
                            unique_keys.remove(record_key)  # 移除已处理的键
                            
                            index_monthly = IndexMonthly(
                                ts_code=row['ts_code'],
                                trade_date=row['trade_date'],
                                open=row['open'],
                                high=row['high'],
                                low=row['low'],
                                close=row['close'],
                                pre_close=row['pre_close'],
                                change=row['change'],
                                pct_chg=row['pct_chg'],
                                vol=row['vol'],
                                amount=row['amount']
                            )
                            
                            if record_key in existing_records:
                                # 更新现有记录
                                existing_record = existing_records[record_key]
                                for field in ['open', 'high', 'low', 'close', 'pre_close', 'change', 
                                            'pct_chg', 'vol', 'amount']:
                                    setattr(existing_record, field, getattr(index_monthly, field))
                                to_update.append(existing_record)
                            else:
                                # 创建新记录
                                to_create.append(index_monthly)
                    
                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        IndexMonthly.objects.bulk_create(to_create, batch_size=1000)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        IndexMonthly.objects.bulk_update(
                            to_update, 
                            ['open', 'high', 'low', 'close', 'pre_close', 'change', 'pct_chg', 
                             'vol', 'amount'],
                            batch_size=1000
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建和更新
            count = await bulk_create_index_monthly(df)
            return count
            
        except Exception as e:
            logger.error(f"Error updating batch index monthly data: {str(e)}")
            raise

    async def batch_update_index_monthly(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新所有指数的月线数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 更新的记录数
        """
        logger.info(f'Starting batch update of all index monthly data from {start_date} to {end_date}')
        total_count = 0
        try:
            # 使用常量中定义的主要指数列表
            ts_codes = [index['ts_code'] for index in MAIN_INDEXES]
            
            # 解析日期
            try:
                if start_date and end_date:
                    if '-' in start_date:  # 格式为YYYY-MM-DD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                    
                    if '-' in end_date:  # 格式为YYYY-MM-DD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
                else:
                    # 如果没有指定日期范围，默认更新最近1年的数据
                    end_date_obj = dt.date.today()
                    start_date_obj = end_date_obj - dt.timedelta(days=365)  # 约1年
            except Exception as e:
                logger.error(f"Error parsing dates: {str(e)}")
                raise
            
            # 将日期对象转换为字符串格式
            start_date_str = start_date_obj.strftime('%Y%m%d')
            end_date_str = end_date_obj.strftime('%Y%m%d')
            
            # 逐个处理每个指数
            total_indexes = len(ts_codes)
            for index_num, ts_code in enumerate(ts_codes, 1):
                # 发送进度更新
                if websocket:
                    message = f"正在更新指数月线数据 ({index_num}/{total_indexes}): {ts_code}"
                    try:
                        await websocket.send_progress(
                            current=index_num,
                            total=total_indexes,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")
                
                try:
                    # 更新单个指数的数据
                    count = await self.update_batch_index_monthly(
                        ts_code=ts_code,
                        start_date=start_date_str,
                        end_date=end_date_str
                    )
                    total_count += count
                    logger.info(f"Updated {count} monthly records for index {ts_code}")
                except Exception as e:
                    logger.error(f"Error updating monthly data for index {ts_code}: {str(e)}")
                    continue
            
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['INDEX_MONTHLY'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {total_count} index monthly records in total")
            return total_count
            
        except Exception as e:
            logger.error(f"Error in batch update of monthly data: {str(e)}")
            raise

    
    ###########################################################################################
    '''
    2025年8月30日增加，更新沪深股通交易前20数据
    '''
    async def update_hsgt_top20(self, start_date=None, end_date=None):
        """
        更新沪深股通交易前20数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :return: 更新的记录数
        """
        logger.info(f'Starting update of HSGT top20 data from {start_date} to {end_date}')
        try:
            # 解析日期
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近10个交易日
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日

            # 获取交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            # 初始化DataFrame
            df_list = []

            # 遍历每个交易日，获取沪深股通前20数据
            call_count = 0
            last_call_time = time.time()
            
            for trade_date in trade_dates:
                # 检查调用频率
                call_count += 1
                if call_count % TUSHARE_API_LIMIT['hsgt_top20_max_in_minute'] == 0:
                    elapsed_time = time.time() - last_call_time
                    if elapsed_time < SECONDS_IN_MINUTE:  # 如果不到一分钟
                        await asyncio.sleep(SECONDS_IN_MINUTE - elapsed_time)
                    last_call_time = time.time()
                
                df = self.pro.hsgt_top10(trade_date=trade_date,)
                if not df.empty:
                    df['trade_date'] = trade_date
                    df_list.append(df)

            if not df_list:
                logger.warning("No HSGT top20 data found in the specified date range")
                return 0

            # 合并所有DataFrame
            final_df = pd.concat(df_list, ignore_index=True)

            # 批量写入数据库
            @sync_to_async
            def bulk_create_hsgt_data(df):
                with transaction.atomic():
                    # 获取现有的记录键值对
                    existing_records = {
                        (record.ts_code, record.trade_date): record 
                        for record in StockHSGT20.objects.filter(
                            trade_date__in=df['trade_date'].unique()
                        )
                    }
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        record_key = (row['ts_code'], row['trade_date'])
                        hsgt_data = StockHSGT20(
                            trade_date=row['trade_date'],
                            ts_code=row['ts_code'],
                            name=row['name'],
                            close=row['close'],
                            change=row['change'],
                            rank=row['rank'],
                            market_type=row['market_type'],
                            amount=row['amount'],
                            net_amount=row['net_amount'],
                            buy=row['buy'],
                            sell=row['sell']
                        )
                        
                        if record_key in existing_records:
                            # 更新现有记录
                            existing_record = existing_records[record_key]
                            existing_record.close = row['close']
                            existing_record.change = row['change']
                            existing_record.rank = row['rank']
                            existing_record.market_type = row['market_type']
                            existing_record.amount = row['amount']
                            existing_record.net_amount = row['net_amount']
                            existing_record.buy = row['buy']
                            existing_record.sell = row['sell']
                            to_update.append(existing_record)
                        else:
                            # 创建新记录
                            to_create.append(hsgt_data)
                    
                    # 批量创建新记录
                    created_count = 0
                    if to_create:
                        StockHSGT20.objects.bulk_create(to_create, batch_size=1000)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        StockHSGT20.objects.bulk_update(
                            to_update, 
                            ['close', 'change', 'rank', 'market_type', 'amount', 'net_amount', 'buy', 'sell'],
                            batch_size=1000
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建
            count = await bulk_create_hsgt_data(final_df)
            
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['HSGT_TOP20'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {count} HSGT top20 records")
            return count
            
        except Exception as e:
            logger.error(f"Error updating HSGT top20 data: {str(e)}")
            raise
    async def update_all_data(self, start_date=None, end_date=None, websocket=None):
        """
        更新全部市场数据
        :param start_date: 开始日期，格式：YYYY-MM-DD
        :param end_date: 结束日期，格式：YYYY-MM-DD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 包含各数据类型更新结果的字典
        """
        results = {}
        try:
            #但实际更新还是看数据库中存储的各类数据的更新日期
            logger.info(f"Starting update of all market data from {start_date} to {end_date}")
            
            # 首先检查是否存在交易日历数据
            trade_cal_count = await sync_to_async(TradeDate.objects.count)()
            if trade_cal_count == 0:
                logger.info("No trade calendar data found, updating trade calendar first...")
                if websocket:
                    await websocket.send_progress(
                        current=0,
                        total=1,#总共多少类数据，后续更新
                        message="正在初始化交易日历数据..."
                    )
                
                # 更新从初始日期到当年12月31日的交易日历
                current_year = dt.date.today().year
                start_date_cal = START_KDATA_DATE.strftime('%Y%m%d') if isinstance(START_KDATA_DATE, dt.date) else '20200102'
                end_date_cal = f"{current_year}1231"
                
                results['trade_cal'] = await self.update_trade_calendar(
                    start_date=start_date_cal,
                    end_date=end_date_cal
                )
                logger.info(f"Initialized trade calendar from {start_date_cal} to {end_date_cal}")
            
            # 转换日期为datetime.date对象，用于计算交易日
            start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date() if start_date else dt.date.today()
            end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date() if end_date else dt.date.today()
            
            #################################首先计算那些数据需要更新###############################################
            #1. 检查是否需要更新交易日历（跨年更新）
            trade_cal_last_update = await get_last_update_date_async(DATA_TYPES['TRADE_CAL'])
            current_year = dt.date.today().year
            last_update_year = trade_cal_last_update.year if trade_cal_last_update else 0
            need_update_trade_cal = current_year > last_update_year
            
            # 计算各数据类型的实际起始日期
            def get_actual_start_date(last_update_date):
                if not last_update_date:
                    return start_date_obj
                else:#以数据库中记录的各个表格时间为准
                    next_day = last_update_date + dt.timedelta(days=1)
                    return  next_day
                
            
            #2.检查个股基本信息是否需要更新（每日更新）
            stock_basic_last_update = await get_last_update_date_async(DATA_TYPES['STOCK_BASIC'])
            stock_basic_start_date = get_actual_start_date(stock_basic_last_update)
            stock_basic_trade_days = await count_trading_days_async(stock_basic_start_date, end_date_obj)
            
            #3. 检查日线数据是否需要更新（每日更新）
            daily_last_update = await get_last_update_date_async(DATA_TYPES['DAILY'])
            daily_start_date = get_actual_start_date(daily_last_update)
            daily_trade_days = await count_trading_days_async(daily_start_date, end_date_obj)
            
            #4. 检查每日指标数据是否需要更新（每日更新）
            daily_basic_last_update = await get_last_update_date_async(DATA_TYPES['DAILY_BASIC'])
            daily_basic_start_date = get_actual_start_date(daily_basic_last_update)
            daily_basic_trade_days = await count_trading_days_async(daily_basic_start_date, end_date_obj)
            print(f'need_update_trade_cal = {need_update_trade_cal} stock_basic_trade_days = {stock_basic_trade_days} \
                  daily_trade_days = {daily_trade_days} daily_basic_trade_days = {daily_basic_trade_days}')
            
            #5. 检查指数日线数据是否需要更新（每日更新）
            index_daily_last_update = await get_last_update_date_async(DATA_TYPES['INDEX_DAILY'])
            index_daily_start_date = get_actual_start_date(index_daily_last_update)
            index_daily_trade_days = await count_trading_days_async(index_daily_start_date, end_date_obj)
            
            #6. 检查指数周线数据是否需要更新（每周更新）
            index_weekly_last_update = await get_last_update_date_async(DATA_TYPES['INDEX_WEEKLY'])
            index_weekly_start_date = get_actual_start_date(index_weekly_last_update)
            index_weekly_trade_days = await count_trading_days_async(index_weekly_start_date, end_date_obj)
            
            #7. 检查指数月线数据是否需要更新（每月更新）
            index_monthly_last_update = await get_last_update_date_async(DATA_TYPES['INDEX_MONTHLY'])
            index_monthly_start_date = get_actual_start_date(index_monthly_last_update)
            index_monthly_trade_days = await count_trading_days_async(index_monthly_start_date, end_date_obj)

            #8检查st股票列表是否需要更新（每日更新）
            st_list_last_update = await get_last_update_date_async(DATA_TYPES['STOCK_ST'])
            st_list_start_date = get_actual_start_date(st_list_last_update)
            st_stock_trade_days = await count_trading_days_async(st_list_start_date, end_date_obj)

            #9检查IPO数据是否需要更新（每日更新）
            ipo_list_last_update = await get_last_update_date_async(DATA_TYPES['IPO_NEW'])
            ipo_list_start_date = get_actual_start_date(ipo_list_last_update)
            ipo_trade_days = await count_trading_days_async(ipo_list_start_date, end_date_obj)

            #10检查个股周K线数据是否需要更新（每日更新）
            stock_weekly_last_update = await get_last_update_date_async(DATA_TYPES['STOCK_WEEKLY'])
            stock_weekly_start_date = get_actual_start_date(stock_weekly_last_update)
            stock_weekly_trade_days = await count_trading_days_async(stock_weekly_start_date, end_date_obj)

            #11检查个股月K线数据是否需要更新（每日更新）
            stock_monthly_last_update = await get_last_update_date_async(DATA_TYPES['STOCK_MONTHLY'])
            stock_monthly_start_date = get_actual_start_date(stock_monthly_last_update)
            stock_monthly_trade_days = await count_trading_days_async(stock_monthly_start_date, end_date_obj)

            #12检查沪深股通top20数据是否需要更新（每日更新）
            hsgt_top20_last_update = await get_last_update_date_async(DATA_TYPES['HSGT_TOP20'])
            hsgt_top20_start_date = get_actual_start_date(hsgt_top20_last_update)
            hsgt_top20_trade_days = await count_trading_days_async(hsgt_top20_start_date, end_date_obj)

            # 设置总步骤数（包括同花顺相关数据和指数数据的更新）
            total_steps = (1 if need_update_trade_cal else 0) + (1 if stock_basic_trade_days > 0 else 0) \
                + (1 if daily_trade_days > 0 else 0) \
                + (1 if daily_basic_trade_days > 0 else 0) \
                + (1 if index_weekly_trade_days > 0 else 0) \
                + (1 if index_monthly_trade_days > 0 else 0) \
                + (1 if index_daily_trade_days > 0 else 0) \
                + (1 if st_stock_trade_days > 0 else 0) \
                + (1 if ipo_trade_days > 0 else 0) \
                + (1 if stock_weekly_trade_days > 0 else 0) \
                + (1 if stock_monthly_trade_days > 0 else 0) \
                + (1 if hsgt_top20_trade_days > 0 else 0)
            current_step = 0
            if websocket:
                await websocket.send_progress(
                    current=0,
                    total=total_steps,
                    message="开始下载数据..."
                )
            ########################################开始正式下载数据##############################################
            # 1.如果需要跨年更新交易日历
            if need_update_trade_cal:
                logger.info(f"Year changed from {last_update_year} to {current_year}, updating trade calendar...")
                
                # 更新从初始日期到当年12月31日的交易日历
                start_date_cal = f"{current_year}0101"
                end_date_cal = f"{current_year}1231"
                
                results['trade_cal'] = await self.update_trade_calendar(
                    start_date=start_date_cal,
                    end_date=end_date_cal
                )
                logger.info(f"Updated trade calendar from {start_date_cal} to {end_date_cal}")
            else:
                results['trade_cal'] = 0
                logger.info(f"Trade calendar is up to date for year {current_year}")
            
            # 2.更新股票基本信息（按日更新）
            if stock_basic_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message="开始更新股票基本信息..."
                    )
                
                results['stock_basic'] = await self.update_stock_basic()
                logger.info("Updated stock basic info")
            
            # 3.批量更新日线数据（只在需要时更新）
            # 获取所有股票，给相关的数据更新共用
            stocks = await get_all_stocks()
            if daily_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新日线数据... (距上次更新: {daily_trade_days}个交易日)"
                    )

                logger.info(f"Updating daily data for {len(stocks)} stocks over {daily_trade_days} trading days")
                
                # 使用新的批处理机制更新日线数据
                results['daily'] = await self.batch_update_daily_data(
                    start_date=daily_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated daily data for {daily_trade_days} trading days from {daily_start_date}")
                
            # 4.批量更新每日指标数据（按日更新，与日K新数据分开更新，但是读取时可能是一起读并合并输出）   
            if daily_basic_trade_days > 0:  
                # 更新每日指标数据
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新每日指标数据... (距上次更新: {daily_basic_trade_days}个交易日)"
                    )
                
                logger.info(f"Updating daily basic data for {len(stocks)} stocks over {daily_basic_trade_days} trading days from {daily_basic_start_date}")
                
                # 使用批处理机制更新每日指标数据
                results['daily_basic'] = await self.batch_update_daily_basic(
                    start_date=daily_basic_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated daily basic data for {daily_basic_trade_days} trading days from {daily_basic_start_date}")
            else:
                results['daily'] = 0
                results['daily_basic'] = 0
                logger.info("No need to update daily data and daily basic data")
            
                 
            # 5.更新指数日线数据
            if index_daily_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新指数日线数据... (距上次更新: {index_daily_trade_days}个交易日)"
                    )
                
                # 使用批处理机制更新指数日线数据
                results['index_daily'] = await self.batch_update_index_daily(
                    start_date=index_daily_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated index daily data for {index_daily_trade_days} trading days from {index_daily_start_date}")
            
            # 6.更新指数周线数据
            if index_weekly_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新指数周线数据... (距上次更新: {index_weekly_trade_days}个交易日)"
                    )
                
                # 使用批处理机制更新指数周线数据
                results['index_weekly'] = await self.batch_update_index_weekly(
                    start_date=index_weekly_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated index weekly data for {index_weekly_trade_days} trading days from {index_weekly_start_date}")
            
            # 7.更新指数月线数据
            if index_monthly_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新指数月线数据... (距上次更新: {index_monthly_trade_days}个交易日)"
                    )
                
                # 使用批处理机制更新指数月线数据
                results['index_monthly'] = await self.batch_update_index_monthly(
                    start_date=index_monthly_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated index monthly data for {index_monthly_trade_days} trading days from {index_monthly_start_date}")

            
            #8 st股票数据
            if st_stock_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新st股票数据... (距上次更新: {st_stock_trade_days}个交易日)"
                        )
                    results['st_stock'] = await self.update_st_stock(
                        start_date=st_list_start_date.strftime('%Y%m%d'),
                        end_date=end_date_obj.strftime('%Y%m%d')
                    )
                  
                    logger.info(f"Updated st stock data for {st_stock_trade_days} trading days from {st_list_start_date}")

            #9 IPO数据
            if ipo_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新IPO数据... (距上次更新: {ipo_trade_days}个交易日)"
                        )
                    results['ipo'] = await self.update_ipo(
                        start_date=ipo_list_start_date.strftime('%Y%m%d'),
                        end_date=end_date_obj.strftime('%Y%m%d')
                    )
                  
                    logger.info(f"Updated ipo data for {ipo_trade_days} trading days from {ipo_list_start_date}")    
                
            #10 个股周线数据
            if stock_weekly_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新个股周线数据... (距上次更新: {stock_weekly_trade_days}个交易日)"
                        )
                    results['stock_weekly'] = await self.batch_update_stock_weekly(
                        start_date=stock_weekly_start_date.strftime('%Y%m%d'),
                        end_date=end_date_obj.strftime('%Y%m%d'),
                        websocket=websocket
                    )
                  
                    logger.info(f"Updated stock weekly data for {stock_weekly_trade_days} trading days from {stock_weekly_start_date}")
            
            #11 个股月线数据
            if stock_monthly_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新个股月线数据... (距上次更新: {stock_monthly_trade_days}个交易日)"
                        )
                    results['stock_monthly'] = await self.batch_update_stock_monthly(
                        start_date=stock_monthly_start_date.strftime('%Y%m%d'), 
                        end_date=end_date_obj.strftime('%Y%m%d'),       
                        websocket=websocket
                    )
                  
                    logger.info(f"Updated stock monthly data for {stock_monthly_trade_days} trading days from {stock_monthly_start_date}")


            #12 沪深股通top20数据
            if hsgt_top20_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新沪深股通top20数据... (距上次更新: {hsgt_top20_trade_days}个交易日)"
                        )
                    results['hsgt_top20'] = await self.update_hsgt_top20(
                        start_date=hsgt_top20_start_date.strftime('%Y%m%d'),
                        end_date=end_date_obj.strftime('%Y%m%d')
                    )
                  
                    logger.info(f"Updated hsgt top20 data for {hsgt_top20_trade_days} trading days from {hsgt_top20_start_date}")
            # 更新数据更新记录（无论是否有更新都记录）
            await update_data_record(
                DATA_TYPES['ALL_DATA_1'],
                end_date_obj
            )

            if websocket:
                await websocket.send_progress(
                    current=total_steps,
                    total=total_steps,
                    message="所有基本数据数据1更新完成"
                )

            logger.info(f"Completed update of all market data: {results}")
            return results
            
        except Exception as e:
            logger.error(f"Error updating all market data: {str(e)}")
            raise